package com.common.base.utils

import android.os.Looper
import android.util.Log
import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers
import io.reactivex.rxjava3.core.Completable
import io.reactivex.rxjava3.core.Flowable
import io.reactivex.rxjava3.core.Single
import io.reactivex.rxjava3.disposables.Disposable
import io.reactivex.rxjava3.functions.Consumer
import io.reactivex.rxjava3.schedulers.Schedulers
import java.util.concurrent.TimeUnit

/**
 * RxJava操作符助手
 */
object RxHelper {

    /**
     * 是否在主线程
     */
    fun isMainThread(): Boolean {
        return Thread.currentThread() == Looper.getMainLooper().thread
    }

    /**
     * 主线程执行
     */
    fun doOnUiThread(task: () -> Unit): Disposable {
        return Flowable
                .just(task)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({
                    it.invoke()
                }, {
                    LogUtils.e(it)
                })
    }

    /**
     * 子线程执行
     */
    fun doOnIOThread(task: () -> Unit): Disposable {
        return Flowable
                .just(task)
                .observeOn(Schedulers.io())
                .subscribe({
                    it.invoke()
                }, {
                    LogUtils.e(it)
                })
    }

    /**
     * 子线程执行
     */
    fun doOnIOThread(task: () -> Unit, onComplete: (() -> Unit)? = null): Disposable {
        return Completable.fromCallable(task)
            .subscribeOn(Schedulers.io())
            .doOnComplete { onComplete?.invoke() }
            .subscribe({}, {
                LogUtils.e(it)
            })
    }

    /**
     * 新线程执行
     */
    fun doOnNewThread(task: () -> Unit): Disposable {
        return Flowable
                .just(task)
                .observeOn(Schedulers.newThread())
                .subscribe({
                    it.invoke()
                }, {
                    LogUtils.e(it)
                })
    }

    /**
     * 新线程执行
     */
    fun doOnNewThread(task: () -> Unit, onComplete: (() -> Unit)? = null): Disposable {
        return Completable.fromCallable(task)
            .subscribeOn(Schedulers.newThread())
            .doOnComplete { onComplete?.invoke() }
            .subscribe({}, {
                LogUtils.e(it)
            })
    }

    /**
     * 新线程执行并返回结果
     */
    fun <T : Any> doOnNewThreadWithResult(task: () -> T, onComplete: ((T) -> Unit)? = null): Disposable {
        return Single.fromCallable(task)
            .subscribeOn(Schedulers.newThread())
            .observeOn(AndroidSchedulers.mainThread()) // 确保在主线程中处理结果
            .subscribe({ result ->
                // 这里返回任务执行的结果
                onComplete?.invoke(result)
            }, { error ->
                // 处理可能发生的错误
                LogUtils.e(error)
            })
    }

    /**
     * 延时执行，在主线程
     */
    fun doDelay(time: Long, task: () -> Unit): Disposable {
        return Flowable.just(task)
                .delay(time, TimeUnit.MILLISECONDS)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({
                    it.invoke()
                }, {
                    LogUtils.e(it)
                })
    }


    /**
     * 延时执行，在IO子线程
     */
    fun doDelayOnIOThread(time: Long, task: () -> Unit): Disposable {
        return Flowable.just(task)
            .delay(time, TimeUnit.MILLISECONDS)
            .observeOn(Schedulers.io())
            .subscribe({
                it.invoke()
            }, {
                LogUtils.e(it)
            })
    }

    /**
     * 轮询操作
     * initialDelay  初始延迟
     * interval      轮询间期
     * unit          轮询间期时间单位
     * consumer      监听事件
     * errorConsumer 出错的事件
     *
     */
    fun polling(initialDelay:Long, interval:Long, unit:TimeUnit,consumer:Consumer<Long>, errorConsumer: Consumer<Throwable> = SimpleErrorConsumer()):Disposable{
        return Flowable.interval(initialDelay, interval, unit)
                .observeOn(Schedulers.newThread())  //回调在子线程中
                .subscribe(consumer, errorConsumer)
    }


    /***
     * 简单错误处理, 不处理会闪退
     */
    class SimpleErrorConsumer: Consumer<Throwable> {

        override fun accept(t: Throwable) {
            Log.i("SimpleErrorConsumer","轮询报错")
        }
    }
}
